package it.learn.kafka.producer81.service;

import org.apache.kafka.clients.admin.*;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

// 简单测试
@Service
public class Test01Service {

    @Autowired
    private ObjectFactory<KafkaAdmin> kafkaAdmin;

    // @PostConstruct
    public void test01() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminClient.create(kafkaAdmin.getObject().getConfigurationProperties());

        // 消费组
        ListConsumerGroupsResult consumerGroups = adminClient.listConsumerGroups();
        Collection<ConsumerGroupListing> groupListings = consumerGroups.all().get();
        String collect = groupListings.stream().map(ConsumerGroupListing::groupId).collect(Collectors.joining(", "));
        System.out.println("一共的消费组有：===> " + collect);

        ListTopicsResult listTopicsResult = adminClient.listTopics();
        Set<String> strings = listTopicsResult.names().get();
        System.out.println("一共的主题有：===> " + strings);

        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList("topic1"));

        adminClient.close();
    }

}
